package com.example.spider.task;

import com.example.spider.cluster.msg.MsgSender;
import com.example.spider.event.listener.SpiderUrlsManager;
import com.example.spider.spider.Spider;
import com.example.spider.starter.SpiderStarter;
import lombok.extern.slf4j.Slf4j;
import org.shoulder.core.context.AppInfo;
import org.shoulder.core.util.ContextUtils;

import java.util.List;
import java.util.concurrent.ThreadPoolExecutor;

/**
 * 爬虫结果处理器
 *
 * @author lym
 */
@SuppressWarnings("rawtypes")
@Slf4j
public class SpiderManager {

    protected final Spider spider;

    protected final SpiderResultHandlerManager resultHandlerManager;

    protected final SpiderUrlsManager spiderUrlsManager;

    protected final List<SpiderStarter> spiderStarters;

    protected final ThreadPoolExecutor spiderThreadPool;

    public SpiderManager(Spider spider, SpiderResultHandlerManager resultHandlerManager, SpiderUrlsManager spiderUrlsManager, List<SpiderStarter> spiderStarters, ThreadPoolExecutor spiderThreadPool) {
        this.spider = spider;
        this.resultHandlerManager = resultHandlerManager;
        this.spiderUrlsManager = spiderUrlsManager;
        this.spiderStarters = spiderStarters;
        this.spiderThreadPool = spiderThreadPool;
    }

    public boolean canAccept() {
        return spiderThreadPool.getQueue().remainingCapacity() > 0;
    }

    public <T> void doSpider(AbstractSpiderTask<T> task) {
        if (task == null) {
            return;
        }
        if (spiderUrlsManager.inNoRetryUrls(task.getUrl())) {
            log.info("drop task for in noRetryUrls taskUrl={}", task.getUrl());
            return;
        }

        log.debug("accept new SpiderTask({})", task.getUrl());
        // 判断本地是否存在幂等文件，使用当前线程，不占用spider
        T idempotentResult = task.calculateIdempotentResult();
        if (idempotentResult != null) {
            resultHandlerManager.handle(task, idempotentResult);
            return;
        }

        if (task.supportCluster() && AppInfo.cluster()) { // fixme 是否已经被分发过，如果是，则不需要再分发
            // 丢到分布式队列，以便各节点均衡
            ContextUtils.getBean(MsgSender.class).addSpiderTask(task, null);
        } else {
            commitSpiderTask(task);
        }
    }

    protected <T> void commitSpiderTask(AbstractSpiderTask<T> task) {
        spiderThreadPool.execute(() -> {
            T result = spider.doSpider(task);
            resultHandlerManager.handle(task, result);
        });
    }

}
